package com.df.ez_game.server;

import com.df.ez_game.server.broker.BrokerMsg;
import com.df.ez_game.server.msg.client.ClientReqMsg;
import com.df.ez_game.server.msg.client.ClientRespMsg;
import com.df.ez_game.server.gateway.UserSession;
import com.df.ez_game.tools.ThreadHelper;
import io.netty.channel.ChannelHandlerContext;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class BrokerManager {
    // 单例实例
    private static BrokerManager instance = new BrokerManager();
    
    /**
     * 获取BrokerManager单例实例
     * @return BrokerManager实例
     */
    public static BrokerManager getInstance() {
        return instance;
    }
    
    // 私有化构造函数
    private BrokerManager() {
        startHeartbeatCheck();
    }
    //broker列表
    List<BrokerInfo> brokers = new ArrayList<>();
    // 心跳检测间隔（毫秒）
    private static final long HEARTBEAT_INTERVAL = 5000;
    // 心跳超时时间（毫秒）
    private static final long HEARTBEAT_TIMEOUT = 15000;

    //broker负载均衡器
    BrokerBalancer balancer = new BrokerBalancer();

    // 构造函数已私有化

    /**
     * 注册Broker节点
     * @param brokerInfo 要注册的Broker节点
     */
    public void registerBroker(BrokerInfo brokerInfo) {
        if (brokerInfo == null) {
            return;
        }
        synchronized (brokers) {
            if (!brokers.contains(brokerInfo)) {
                brokers.add(brokerInfo);
                brokerInfo.connect();
            }
        }
    }

    /**
     * 注销Broker节点
     * @param brokerInfo 要注销的Broker节点
     */
    public void unregisterBroker(BrokerInfo brokerInfo) {
        if (brokerInfo == null) {
            return;
        }
        synchronized (brokers) {
            brokers.remove(brokerInfo);
        }
    }

    /**
     * 启动心跳检测定时任务
     */
    private void startHeartbeatCheck() {
        ThreadHelper.scheduleAtFixedRate(() -> {
            synchronized (brokers) {
                Iterator<BrokerInfo> iterator = brokers.iterator();
                while (iterator.hasNext()) {
                    BrokerInfo broker = iterator.next();
                    if (!broker.checkAlive(HEARTBEAT_TIMEOUT)) {
                        // 节点不存活，从列表中移除
                        iterator.remove();
                        System.out.println("Broker节点下线: " + broker);
                    }
                }
            }
        }, HEARTBEAT_INTERVAL, HEARTBEAT_INTERVAL, TimeUnit.MILLISECONDS);
    }

    /**
     * 处理Broker心跳
     * @param brokerInfo 发送心跳的Broker节点
     */
    public void processHeartbeat(BrokerInfo brokerInfo) {
        if (brokerInfo == null) {
            return;
        }
        brokerInfo.updateHeartbeat();
    }

    /**
     * 消息转发到Broker节点
     * @param ctx 通道上下文
     * @param gateWayReqMsg 网关请求消息
     * @param session 用户会话
     */
    public void forwarding(ChannelHandlerContext ctx, ClientReqMsg gateWayReqMsg, UserSession session) {
        boolean needResp = gateWayReqMsg.getNeedResp();
        // 消息转换为内部消息
        BrokerMsg brokerReqMsg = toBrokerReqMsg(gateWayReqMsg, session);
        // 只选择存活的Broker节点
        List<BrokerInfo> aliveBrokers = getAliveBrokers();
        
        if (aliveBrokers.isEmpty()) {
            if(needResp){
                sendErrorResponse(ctx, gateWayReqMsg, "无可用Broker节点");
            }
            return;
        }
        
        try {
            // 选择一个服务器进行发送
            BrokerInfo brokerInfo = balancer.selectBroker(aliveBrokers);
            brokerInfo.sendMsg(ctx, brokerReqMsg);
        } catch (Exception e) {
            if(needResp){
                sendErrorResponse(ctx, gateWayReqMsg, "服务调用异常: " + e.getMessage());
            }
        }
    }

    /**
     * 发送错误响应
     * @param ctx 通道上下文
     * @param reqMsg 原始请求消息
     * @param error 错误信息
     */
    private void sendErrorResponse(ChannelHandlerContext ctx, ClientReqMsg reqMsg, String error) {
        ClientRespMsg respMsg = new ClientRespMsg();
        respMsg.setCode(-1);
        respMsg.setCmd(reqMsg.getCmd());
//        respMsg.setError(error);
        ctx.channel().writeAndFlush(respMsg);
    }

    /**
     * 获取存活的Broker节点列表
     * @return 存活的Broker节点列表
     */
    private List<BrokerInfo> getAliveBrokers() {
        List<BrokerInfo> aliveBrokers = new ArrayList<>();
        synchronized (brokers) {
            for (BrokerInfo broker : brokers) {
                if (broker.checkAlive(HEARTBEAT_TIMEOUT)) {
                    aliveBrokers.add(broker);
                }
            }
        }
        return aliveBrokers;
    }

    BrokerMsg toBrokerReqMsg(ClientReqMsg gateWayReqMsg, UserSession session){
        //携带发送服的id ；携带绑定信息
        BrokerMsg brokerMsg = new BrokerMsg();
        brokerMsg.setMsgType(1);
        brokerMsg.setFrom(1);
        brokerMsg.setTo(1);
        brokerMsg.setVer(gateWayReqMsg.getVer());
        brokerMsg.setCmd(gateWayReqMsg.getCmd());
        brokerMsg.setMsgTag(gateWayReqMsg.getMsgSerial());
        brokerMsg.setData(gateWayReqMsg.getData());
        brokerMsg.setNeedResp(gateWayReqMsg.getNeedResp());
        brokerMsg.getBind().addAll(session.getLogicBind());
        return brokerMsg;

    }
}
